//external sort 

package qp.operators;

import java.io.*;

import qp.utils.*;
import qp.optimizer.*;
import qp.parser.*;
import qp.operators.*;

import java.util.*;

public class ExternalSort
{
	Schema sortSchema; //contain the index of tuple
	
	int fileIndex; //get a unique file name 

	//temerary variable used in sort()
	Batch page;
	ObjectInputStream in; 
	ObjectOutputStream out;
	
	int bufferNum;// buffer size used for sorting, assigned by buffer manager
	int batchSize;
	int tupleSize;
	String finalFile;  //maitian the sorted file

	//attribute list
	Vector<Attribute> attrList;
	Vector<Integer> attrListIndex = new Vector<Integer>();

	//the index used to build unique file name used in external sort
	static int filenum = 0;

	//the queue to maitain the file in external sort
	LinkedList<String> sortQueue; 
	
	public ExternalSort(String inputFile, Vector<Attribute> attribuList, Schema sche)
	{
		//input stream
		sortSchema = sche;
		
		try{
			 in = new ObjectInputStream(new FileInputStream(inputFile));
		}catch(IOException io)
		{
		    System.err.println("External::sort error in reading the file");
		    System.exit(1);
		}

		//condition attribute list
		attrList = attribuList;
		
		for(int i=0;i<attrList.size();i++) {
//			Debug.PPrint(attrList.elementAt(i));
			int index=sortSchema.indexOf(attrList.elementAt(i));
			attrListIndex.add(index);
//			System.out.println("sort on attribute:::"+index);
		}
		
		//buffer size 
		bufferNum=BufferManager.getBufferForSort();
		if(bufferNum<3) {
			System.err.println("buffer size for external sort at least 3 pages");
			System.exit(1);
		}
		tupleSize=sortSchema.getTupleSize();
		batchSize=Batch.getPageSize()/tupleSize;
		//build the queue
		sortQueue = new LinkedList<String>();
	}

	//create a unique file name
	private String getUniqueName()
	{
			filenum++; 
			String tempName;
			tempName = "ExternalSort_" +  String.valueOf(filenum) + ".data";

			return tempName;
	}

	public String sort() throws IOException, ClassNotFoundException {
		
		Block inputBlock=new Block(bufferNum-1);
		boolean eof=false;//end of the input file
		Vector<String> readFiles=new Vector<String>();
		Vector<String> writeFiles=new Vector<String>();
		Batch outputBatch=new Batch(batchSize);
		Vector<ObjectInputStream> blockois;
		String outputfilename=null;
		ObjectOutputStream oos=null;
		int min=0;
		//****phase 1****
		//we do not use all the buffer for phase, 
		//only use bufferSize-1, since we are using merge sort in phase 1
		while(!eof) {
			inputBlock=new Block(bufferNum-1);
			outputfilename=getUniqueName();
			//load batches into buffer
			for(int i=0;i<bufferNum-1;i++) {
				try {
					inputBlock.add(i, (Batch)in.readObject());
				}
				catch (EOFException e) {
					// TODO: handle exception
					eof=true;
					break;
				}
				catch (Exception e) {
					// TODO: handle exception
				}
			}

			if(inputBlock.size()>0) {
				//locally sort all the buffer
				for(int i=0;i<inputBlock.size();i++) {
					inputBlock.batchAt(i).sort(attrListIndex.firstElement());
				}
				try {
					FileOutputStream fos = new FileOutputStream(outputfilename);
					oos=new ObjectOutputStream(fos);
					while(inputBlock.existTupleInBlock>0) {
						min=inputBlock.findMinFromSortedBatch(attrListIndex.firstElement());
						outputBatch.add(inputBlock.batchAt(min).elementAt(0));
						
						/*for debug
						System.out.println("=====Before deleting===");
						for(int i=0;i<inputBlock.size();i++) {
							System.out.println("========");
							Debug.PPrint(inputBlock.batchAt(i));
						}
						System.out.println("=====Min tuple===");
						Debug.PPrint(inputBlock.batchAt(min).elementAt(0));
						Thread.sleep(1000);
						*///end debug
						
						inputBlock.batchAt(min).remove(0);
						inputBlock.existTupleInBlock--;
						
						//if the outputBatch is full, write it to disk and reset the outputbatch
						if(outputBatch.isFull()) {
							oos.writeObject(outputBatch);
							outputBatch=new Batch(batchSize);
						}
					}
					if(outputBatch.size()>0) {
						oos.writeObject(outputBatch);
						outputBatch=new Batch(batchSize);
					}
					oos.close();
					fos.close();
					writeFiles.add(outputfilename);
				} 
				catch (FileNotFoundException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				catch (Exception e) {
					// TODO: handle exception
					e.printStackTrace();
				}
			}
			else {
				break;
			}		
		}
		//****end of phase 1****
		
		readFiles=writeFiles;
//		System.out.println("writeFiles size="+writeFiles.size());
		writeFiles=new Vector<String>();
		
		//****phase 2****
		while(readFiles.size()>0) {
			inputBlock=new Block(bufferNum-1);
			blockois=new Vector<ObjectInputStream>();
			oos=null;//stream for output
			//every round,create a new file to stored the output page
			
			outputfilename=getUniqueName();
			try {
				FileOutputStream fos=new FileOutputStream(outputfilename);
				oos=new ObjectOutputStream(fos);
			}
			catch (Exception e) {
				// TODO: handle exception
				System.out.println("error when create output file in phase 2");
			}
			
			for(int i=0;i<bufferNum-1;i++) {
				try {
					FileInputStream fis=new FileInputStream(readFiles.elementAt(0));
					ObjectInputStream ois=new ObjectInputStream(fis);
					blockois.add(ois);
//					System.out.println("inputblock size===="+inputBlock.size());
					inputBlock.add(i, (Batch)ois.readObject());
					System.out.println(readFiles.elementAt(0));
					readFiles.remove(0);
				}
				catch (ArrayIndexOutOfBoundsException e) {
					// TODO: handle exception
//					System.out.println("readFiles.size()======="+readFiles.size());
//					System.out.println("writeFiles.size()======="+writeFiles.size());
//					System.out.println("ArrayIndexOutOfBoundsException in first time load in phase 2");
					break;
				}
				catch (IOException e) {
					// TODO: handle exception
					
				}
				catch (Exception e) {
					// TODO: handle exception
					e.printStackTrace();
				}
			}


			//for debug
			for(int i=0;i<inputBlock.size();i++) {
//				System.out.println("====Out put one batch====And batchsize=="+batchSize);
//				Debug.PPrint(inputBlock.batchAt(i));
			}
			//end debug
			while(inputBlock.existTupleInBlock>0) {
				min=inputBlock.findMinFromSortedBatch(attrListIndex.firstElement());
				outputBatch.add(inputBlock.batchAt(min).elementAt(0));
				inputBlock.batchAt(min).remove(0);
				inputBlock.existTupleInBlock--;

				//if the one of the page is exhausted, fill it from ObjectInputStream
				if(inputBlock.batchAt(min).size()==0) {
					inputBlock.remove(min);
					try {
						Batch newBatch=(Batch)blockois.elementAt(min).readObject();
						inputBlock.add(min, newBatch);
//						System.out.println("===No::"+min+" stream load a new batch ===");
					}
					catch (EOFException e) {
						// TODO: handle exception
						try {
							blockois.elementAt(min).close();
//							System.out.println("===No::"+min+" stream get the end ===");
							blockois.remove(min);
						}
						catch (Exception e1) {
							// TODO: handle exception
						}
					}
					catch (Exception e) {
						// TODO: handle exception
					}
				}
				
				if(outputBatch.isFull()) {
					oos.writeObject(outputBatch);
					outputBatch=new Batch(batchSize);
//					System.out.println("-------write full batch-------");
				}
			}
			if(outputBatch.size()>0) {
				oos.writeObject(outputBatch);
				outputBatch=new Batch(batchSize);
//    			System.out.println("-------write un full batch-------");

			}
			
			//add the file to the vector of writefiles
			writeFiles.add(outputfilename);
			System.out.printf("         ====================>");
			System.out.printf("                       %s\n\n\n", writeFiles.lastElement());
			if(readFiles.size()==0){
				if(writeFiles.size()==1) {
					return writeFiles.firstElement();
				}
				else {
					readFiles=writeFiles;
					writeFiles=new Vector<String>();
				}
			}
		}
		return null;
	}
	
}
